package com.dcloud.streamweb.core.jobserver.impl;

import cn.hutool.core.date.DateUtil;
import com.dcloud.streamweb.constants.MessageConstants;
import com.dcloud.streamweb.constants.SystemConstants;
import com.dcloud.streamweb.core.command.CommandRpcClinetAdapter;
import com.dcloud.streamweb.core.jobserver.JobBaseServiceAO;
import com.dcloud.streamweb.core.rpc.FlinkRestRpcAdapter;
import com.dcloud.streamweb.core.rpc.model.JobStandaloneInfo;
import com.dcloud.streamweb.enums.*;
import com.dcloud.streamweb.exception.BizException;
import com.dcloud.streamweb.log.entity.JobRunLog;
import com.dcloud.streamweb.log.entity.dto.JobRunLogDTO;
import com.dcloud.streamweb.log.service.IJobRunLogService;
import com.dcloud.streamweb.sysconfig.config.CustomConfig;
import com.dcloud.streamweb.sysconfig.config.JobThreadPool;
import com.dcloud.streamweb.sysconfig.entity.dto.SystemConfigDTO;
import com.dcloud.streamweb.sysconfig.service.ISystemConfigService;
import com.dcloud.streamweb.taskmanage.entity.dto.JobConfigDTO;
import com.dcloud.streamweb.taskmanage.entity.dto.JobRunParamDTO;
import com.dcloud.streamweb.taskmanage.service.IJobConfigService;
import com.dcloud.streamweb.util.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author zhuhuipei
 * @Description:
 * @date 2021/3/28
 * @time 10:02
 */
@Component
@Slf4j
public class JobBaseServiceAOImpl implements JobBaseServiceAO {

    public static final ThreadLocal<String> THREADAPPID = new ThreadLocal<String>();

    @Autowired
    private ISystemConfigService systemConfigService;

    @Autowired
    private CustomConfig customConfig;

    @Autowired
    private IJobRunLogService jobRunLogService;

    @Autowired
    private CommandRpcClinetAdapter commandRpcClinetAdapter;

    @Autowired
    private FlinkRestRpcAdapter flinkRestRpcAdapter;

    @Autowired
    private IJobConfigService jobConfigService;

    /**
     * 功能描述 判断任务的状态，如果在运行中，或者执行中，则抛出异常，组织提交任务执行
     *
     * @param jobConfigDTO
     * @return void
     * @author shijf
     * @date 2024/7/25 上午12:04
     */
    @Override
    public void checkStart(JobConfigDTO jobConfigDTO) {


        if (jobConfigDTO == null) {
            throw new BizException(SysErrorEnum.JOB_CONFIG_JOB_IS_NOT_EXIST);
        }
        if (JobConfigStatus.RUN.getCode().equals(jobConfigDTO.getStatus().getCode())) {
            throw new BizException("任务运行中请先停止任务");
        }
        if (jobConfigDTO.getStatus().equals(JobConfigStatus.STARTING)) {
            throw new BizException("任务正在启动中 请稍等..");
        }


        //根据运行模式 判断系统设置参数有没有设置
        Map<String, String> systemConfigMap = SystemConfigDTO
                .toMap(systemConfigService.getSystemConfig(SysConfigEnumType.SYS));
        this.checkSysConfig(systemConfigMap, jobConfigDTO.getDeployModeEnum());
    }

    //将sql文件写入到本地，同时提供出一个访问地址
    @Override
    public JobRunParamDTO writeSqlToFile(JobConfigDTO jobConfigDTO) {

        Map<String, String> systemConfigMap = SystemConfigDTO
                .toMap(systemConfigService.getSystemConfig(SysConfigEnumType.SYS));
        String fileName = FileUtils.createFileName(String.valueOf(jobConfigDTO.getId()));
        String sqlPath = FileUtils
                .getSqlHome(systemConfigMap.get(SysConfigEnum.FLINK_STREAMING_PLATFORM_WEB_HOME.getKey()))
                + fileName;
        FileUtils.writeText(sqlPath, jobConfigDTO.getFlinkSql(), Boolean.FALSE);

        String sqlUrl = customConfig.getUrlForReadLocal() + fileName;

        return JobRunParamDTO.buildJobRunParam(systemConfigMap, jobConfigDTO, sqlPath, sqlUrl);
    }

    @Override
    public Long insertJobRunLog(JobConfigDTO jobConfigDTO, String userName) {
        JobRunLogDTO jobRunLogDTO = new JobRunLogDTO();
        jobRunLogDTO.setDeployMode(jobConfigDTO.getDeployModeEnum().name());
        jobRunLogDTO.setLocalLog(MessageConstants.MESSAGE_001);
        jobRunLogDTO.setJobConfigId(jobConfigDTO.getId());
        jobRunLogDTO.setStartTime(new Date());
        jobRunLogDTO.setJobName(jobConfigDTO.getJobName());
        jobRunLogDTO.setJobId(jobConfigDTO.getJobId());
        jobRunLogDTO.setJobStatus(JobStatusEnum.STARTING.name());
        jobRunLogDTO.setCreator(userName);
        jobRunLogDTO.setEditor(userName);
        jobRunLogDTO.setRunIp(IpUtil.getInstance().getLocalIP());
        JobRunLog entity = JobRunLogDTO.toEntity(jobRunLogDTO);
        jobRunLogService.save(entity);
        return entity.getId();
    }

    @Override
    public void aSyncExecJob(JobRunParamDTO jobRunParamDTO, JobConfigDTO jobConfigDTO, Long jobRunLogId, String savepointPath) {
        ThreadPoolExecutor threadPoolExecutor = JobThreadPool.getInstance().getThreadPoolExecutor();
        threadPoolExecutor.execute(new Runnable() {
            @Override
            public void run() {
                String jobStatus = JobStatusEnum.SUCCESS.name();
                String appId = "";
                THREADAPPID.set(appId);
                boolean success = true;
                StringBuilder localLog = new StringBuilder()
                        .append("开始提交任务：")
                        .append(DateUtil.now()).append(SystemConstants.LINE_FEED)
                        .append("三方jar: ").append(SystemConstants.LINE_FEED);
                if (jobConfigDTO.getExtJarPath() != null) {
                    localLog.append(jobConfigDTO.getExtJarPath());
                }

                localLog.append(SystemConstants.LINE_FEED)
                        .append("客户端IP：").append(IpUtil.getInstance().getLocalIP())
                        .append(SystemConstants.LINE_FEED);
                try {
                    String command = "";
                    jobConfigDTO.setDownloadUrl(customConfig.getUrlForDown());
                    String hadoopClassPath = SystemInfoUtil.getEnv(SystemConstants.HADOOP_CLASSPATH);
                    log.info("HADOOP_CLASSPATH={}", hadoopClassPath);
                    localLog.append("HADOOP_CLASSPATH=").append(hadoopClassPath)
                            .append(SystemConstants.LINE_FEED);
                    //如果是自定义提交jar模式下载文件到本地
                    this.downJar(jobRunParamDTO, jobConfigDTO);
                    switch (jobConfigDTO.getDeployModeEnum()) {
                        case YARN_PER:
                        case YARN_APPLICATION:

                            if (StringUtils.isEmpty(hadoopClassPath)) {
                                throw new BizException("yarn 模式 需要在客户端配置 HADOOP_CLASSPATH环境变量");
                            }
                            log.info("YARN模式下 HADOOP_CLASSPATH={}", hadoopClassPath);

                            //1、构建执行命令
                            command = CommandUtil.buildRunCommandForYarnCluster(jobRunParamDTO,
                                    jobConfigDTO, savepointPath);
                            //2、提交任务
                            appId = this.submitJobForYarn(command, jobConfigDTO, localLog);
                            THREADAPPID.set(appId);
                            break;
                        case LOCAL:
                        case STANDALONE:
                            String address = systemConfigService
                                    .getFlinkAddress(jobConfigDTO.getDeployModeEnum());
                            log.info("flink 远程提交地址是 address={}", address);
                            localLog.append("address=").append(address).append(SystemConstants.LINE_FEED);
                            //1、构建执行命令
                            command = CommandUtil
                                    .buildRunCommandForCluster(jobRunParamDTO, jobConfigDTO, savepointPath, address);
                            //2、提交任务
                            appId = this.submitJobForStandalone(command, jobConfigDTO, localLog);
                            break;
                        default:
                            log.warn("不支持的模式 {}", jobConfigDTO.getDeployModeEnum());
                    }
                } catch (Exception e) {
                    log.error("任务[" + jobConfigDTO.getId() + "]执行有异常！", e);
                    localLog.append(e).append(errorInfoDir());
                    success = false;
                    jobStatus = JobStatusEnum.FAIL.name();
                } finally {
                    localLog.append("\n启动结束时间: ").append(DateUtil.now()).append(SystemConstants.LINE_FEED);
                    if (success) {
                        localLog.append(
                                "######客户端提交任务 成功 （客户端提交成功不代表任务在集群运行成功，具体请查看集群任务）############################## ");
                    } else {
                        localLog.append("######启动结果是 失败############################## ");
                    }
                    if (StringUtils.isBlank(appId)) { // 解决任务异常，但已经生成了appID，但没有传递给上层调用方法的问题
                        appId = THREADAPPID.get();
                        log.info("任务[{}]执行有异常情况，appid = {}", jobConfigDTO.getId(), appId);
                    }

                    //根据flink执行情况，更新本地数据库状态
                    this.updateStatusAndLog(jobConfigDTO, jobRunLogId, jobStatus, localLog.toString(), appId);
                }
            }


            /**
             * 下载文件到本地并且setMainJarPath
             *
             * @author zhuhuipei
             * @date 2021/3/28
             * @time 14:58
             */
            private void downJar(JobRunParamDTO jobRunParamDTO, JobConfigDTO jobConfigDTO) {
                if (JobTypeEnum.JAR.getCode() == jobConfigDTO.getJobTypeEnum().getCode()) {
                    String pathName = null;
                    if (MatcherUtils.isHttpsOrHttp(jobConfigDTO.getCustomJarUrl())) {
                        String savePath = jobRunParamDTO.getSysHome() + "tmp/udf_jar/";
                        try {
                            pathName = UrlUtil.downLoadFromUrl(jobConfigDTO.getCustomJarUrl(), savePath);
                        } catch (Exception e) {
                            log.error("文件下载失败 {} path={}", jobConfigDTO.getCustomJarUrl(), savePath, e);
                            throw new BizException("文件下载失败");
                        }
                    } else {
                        pathName =
                                systemConfigService.getUploadJarsPath() + SystemConstants.VIRGULE + jobConfigDTO
                                        .getCustomJarUrl();
                    }
                    log.info("JAR模式下 pathName={}", pathName);
                    jobRunParamDTO.setMainJarPath(pathName);

                }
            }

            /**
             * 错误日志目录提示
             *
             * @author zhuhuipei
             * @date 2020-10-19
             * @time 21:47
             */
            private String errorInfoDir() {
                StringBuilder errorTips = new StringBuilder(SystemConstants.LINE_FEED)
                        .append(SystemConstants.LINE_FEED)
                        .append("（重要）请登陆服务器分别查看下面两个目录下的错误日志")
                        .append(IpUtil.getInstance().getLocalIP()).append(SystemConstants.LINE_FEED)
                        .append("web系统日志目录（web日志）：")
                        .append(systemConfigService
                                .getSystemConfigByKey(SysConfigEnum.FLINK_STREAMING_PLATFORM_WEB_HOME.getKey()))
                        .append("logs/error.log").append(SystemConstants.LINE_FEED)
                        .append("flink提交日志目录（flink客户端日志：")
                        .append(systemConfigService.getSystemConfigByKey(SysConfigEnum.FLINK_HOME.getKey()))
                        .append("log/)").append(SystemConstants.LINE_FEED)
                        .append(SystemConstants.LINE_FEED)
                        .append(SystemConstants.LINE_FEED);
                return errorTips.toString();
            }


            /**
             * 更新日志、更新配置信息
             *
             * @param jobConfig
             * @param jobRunLogId
             * @param jobStatus
             * @param localLog
             * @param appId
             */
            private void updateStatusAndLog(JobConfigDTO jobConfig, Long jobRunLogId,
                                            String jobStatus, String localLog, String appId) {
                try {
                    JobConfigDTO jobConfigDTO = new JobConfigDTO();
                    jobConfigDTO.setId(jobConfig.getId());
                    jobConfigDTO.setJobId(appId);
                    JobRunLogDTO jobRunLogDTO = new JobRunLogDTO();
                    jobRunLogDTO.setId(jobRunLogId);
                    jobRunLogDTO.setJobId(appId);
                    if (JobStatusEnum.SUCCESS.name().equals(jobStatus) && !StringUtils.isEmpty(appId)) {

                        //批任务提交完成后算成功
                        if (JobTypeEnum.SQL_BATCH.equals(jobConfig.getJobTypeEnum())) {
                            jobConfigDTO.setStatus(JobConfigStatus.SUCCESS);
                        } else {
                            jobConfigDTO.setStatus(JobConfigStatus.RUN);
                        }

                        jobConfigDTO.setLastStartTime(new Date());
                        if (DeployModeEnum.LOCAL.equals(jobConfig.getDeployModeEnum())
                                || DeployModeEnum.STANDALONE.equals(jobConfig.getDeployModeEnum())) {
                            jobRunLogDTO.setRemoteLogUrl(
                                    systemConfigService.getFlinkHttpAddress(jobConfig.getDeployModeEnum())
                                            + SystemConstants.HTTP_STANDALONE_APPS + jobConfigDTO.getJobId());
                        }
                        if (DeployModeEnum.YARN_PER.equals(jobConfig.getDeployModeEnum())) {
                            jobRunLogDTO.setRemoteLogUrl(systemConfigService.getYarnRmHttpAddress()
                                    + SystemConstants.HTTP_YARN_CLUSTER_APPS + jobConfigDTO.getJobId());
                        }

                    } else {
                        jobConfigDTO.setStatus(JobConfigStatus.FAIL);
                    }
                    jobConfigService.updateJobConfigById(jobConfigDTO);

                    jobRunLogDTO.setJobStatus(jobStatus);
                    jobRunLogDTO.setLocalLog(localLog);
                    jobRunLogService.updateJobRunLogById(jobRunLogDTO);

                    //最后更新一次日志 (更新日志和更新信息分开 防止日志更新失败导致相关状态更新也失败)
                    jobRunLogService.updateLogById(localLog, jobRunLogId);
                } catch (Exception e) {
                    log.error(" localLog.length={} 异步更新数据失败：", localLog.length(), e);
                }
            }

            private String submitJobForStandalone(String command, JobConfigDTO jobConfig,
                                                  StringBuilder localLog)
                    throws Exception {
                String appId = commandRpcClinetAdapter
                        .submitJob(command, localLog, jobRunLogId, jobConfig.getDeployModeEnum());
                JobStandaloneInfo jobStandaloneInfo = flinkRestRpcAdapter
                        .getJobInfoForStandaloneByAppId(appId,
                                jobConfig.getDeployModeEnum());

                if (jobStandaloneInfo == null || StringUtils.isNotEmpty(jobStandaloneInfo.getErrors())) {
                    log.error("[submitJobForStandalone] is error jobStandaloneInfo={}", jobStandaloneInfo);
                    localLog.append("\n 任务失败 appId=" + appId);
                    throw new BizException("任务失败");
                } else {
                    if (!SystemConstants.STATUS_RUNNING.equals(jobStandaloneInfo.getState())
                            && !SystemConstants.STATUS_FINISHED.equals(jobStandaloneInfo.getState())) {
                        localLog.append("\n 任务失败 appId=" + appId).append("状态是：" + jobStandaloneInfo.getState());
                        throw new BizException("[submitJobForStandalone]任务失败");
                    }
                }
                return appId;
            }

            //todo yarn 先不做
            private String submitJobForYarn(String command, JobConfigDTO jobConfigDTO,
                                            StringBuilder localLog)
                    throws Exception {
                commandRpcClinetAdapter
                        .submitJob(command, localLog, jobRunLogId, jobConfigDTO.getDeployModeEnum());
//                return yarnRestRpcAdapter.getAppIdByYarn(jobConfigDTO.getJobName(),
//                        YarnUtil.getQueueName(jobConfigDTO.getFlinkRunConfig()));
                return null;
            }

        });
    }

    /**
     * @Description: 判断该任务有没有设置检查点
     * @param jobConfigDTO
     * @return: void
     * @Author: shijf
     * @Date 2024/7/25 16:40
     */
    @Override
    public void checkSavepoint(JobConfigDTO jobConfigDTO) {


        if (jobConfigDTO == null) {
            throw new BizException(SysErrorEnum.JOB_CONFIG_JOB_IS_NOT_EXIST);
        }
//        if (JobTypeEnum.JAR.equals(jobConfigDTO.getJobTypeEnum())) {
//            log.warn(MessageConstants.MESSAGE_006, jobConfigDTO.getJobName());
//            throw new BizException(MessageConstants.MESSAGE_006);
//        }
        if (JobTypeEnum.SQL_BATCH.equals(jobConfigDTO.getJobTypeEnum())) {
            log.warn(MessageConstants.MESSAGE_010, jobConfigDTO.getJobName());
            throw new BizException(MessageConstants.MESSAGE_010);
        }

        if (StringUtils.isEmpty(jobConfigDTO.getFlinkCheckpointConfig())
                && DeployModeEnum.STANDALONE.equals(jobConfigDTO.getDeployModeEnum())) {
            log.error(MessageConstants.MESSAGE_004, jobConfigDTO);
            throw new BizException(MessageConstants.MESSAGE_004);
        }
        if (StringUtils.isEmpty(jobConfigDTO.getJobId())) {
            log.warn(MessageConstants.MESSAGE_005, jobConfigDTO.getJobName());
            throw new BizException(MessageConstants.MESSAGE_005);
        }
    }


    private void checkSysConfig(Map<String, String> systemConfigMap, DeployModeEnum deployModeEnum) {
        if (systemConfigMap == null) {
            throw new BizException(SysErrorEnum.SYSTEM_CONFIG_IS_NULL);
        }

        //flink安装目录必填
        if (!systemConfigMap.containsKey(SysConfigEnum.FLINK_HOME.getKey())) {
            throw new BizException(SysErrorEnum.SYSTEM_CONFIG_IS_NULL_FLINK_HOME);
        }

        if (DeployModeEnum.YARN_PER.equals(deployModeEnum) || DeployModeEnum.YARN_APPLICATION
                .equals(deployModeEnum)) {
            if (!systemConfigMap.containsKey(SysConfigEnum.YARN_RM_HTTP_ADDRESS.getKey())) {
                throw new BizException(SysErrorEnum.SYSTEM_CONFIG_IS_NULL_YARN_RM_HTTP_ADDRESS);
            }
        }
        if (!systemConfigMap.containsKey(SysConfigEnum.FLINK_STREAMING_PLATFORM_WEB_HOME.getKey())) {
            throw new BizException(SysErrorEnum.SYSTEM_CONFIG_IS_NULL_FLINK_STREAMING_PLATFORM_WEB_HOME);
        }
    }
}
